package org.future.task.redis.connection;

import org.future.task.core.handler.CacheConnection;
import org.future.task.core.handler.Commit;
import org.future.task.core.serialize.MineSerializer;
import org.future.task.redis.commit.RedisListCommit;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;

public class RedisConnectionImpl implements CacheConnection {

    private RedisTemplate redisTemplate;

    private RedisListCommit redisListCommit;

    private MineSerializer<byte[], Object> mineSerializer;

    private volatile boolean hasVal;

    public RedisConnectionImpl(RedisTemplate redisTemplate) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(redisTemplate, null, 0);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, int writeBatch) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(redisTemplate, null, writeBatch);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, MineSerializer<byte[], Object> mineSerializer) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        this(redisTemplate, mineSerializer, 0);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, MineSerializer<byte[], Object> mineSerializer, int writeBatch) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        this(redisTemplate, mineSerializer, writeBatch, 0, 0);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, int writeBatch, int localCacheSize) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        this(redisTemplate, null, writeBatch, localCacheSize, 0);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, MineSerializer<byte[], Object> mineSerializer, int writeBatch, int commitThreadSize) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        this(redisTemplate, mineSerializer, writeBatch, 0, 0);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, int writeBatch, int localCacheSize, int commitThreadSize) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        this(redisTemplate, null, writeBatch, localCacheSize, 0);
    }

    public RedisConnectionImpl(RedisTemplate redisTemplate, MineSerializer<byte[], Object> mineSerializer, int writeBatch, int localCacheSize, int commitThreadSize) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        this.redisTemplate = redisTemplate;
        if (localCacheSize <= 0) {
            localCacheSize = 1000;
        }
        redisListCommit = new RedisListCommit(this, localCacheSize, commitThreadSize);
        redisListCommit.setWriteBatch(writeBatch);
        if (mineSerializer == null) {
            this.mineSerializer = (MineSerializer<byte[], Object>) Class.forName("org.future.task.redis.serialize.ByteMineSerializer").newInstance();
        }
    }


    @Override
    public List<Object> getDataList(final String queueName, int size, int timeOut) {
        if (!hasVal) {
            RedisConnection connection = null;
            List<byte[]> bytes = null;
            try {
                connection = Objects.requireNonNull(redisTemplate.getConnectionFactory()).getConnection();
                bytes = connection.bLPop(timeOut < 0 ? 300 : timeOut, queueName.getBytes());

                if (bytes == null || bytes.isEmpty() || bytes.size() < 2) {
                    return null;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (connection != null) {
                    connection.close();
                }
            }
            hasVal = true;

            return Collections.singletonList(mineSerializer.decode(bytes.get(1)));
        }
        final List<Object> resultList = new ArrayList<>();

        redisTemplate.executePipelined((RedisConnection redisConnection) -> {
            byte[] serialize = queueName.getBytes();
            for (int i = 0; i < size; i++) {
                redisConnection.lPop(serialize);
            }
            List<Object> objs = redisConnection.closePipeline();
            objs.removeAll(Collections.singleton(null));
            for (Object obj : objs) {
                resultList.add(mineSerializer.decode((byte[]) obj));
            }
            return null;
        });

        if (resultList.isEmpty()) {
            hasVal = false;
            return null;
        }
        return resultList;
    }

    @Override
    public long size(String queueName) {
        return redisTemplate.boundListOps(queueName).size();
    }


    @Override
    public void sendDataList(String queueName, List<?> dataList) throws Exception {
        if (queueName == null || dataList == null || dataList.isEmpty()) {
            return;
        }
        redisTemplate.executePipelined(new RedisCallback<Object>() {

            @Override
            public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
                byte[] serialize = queueName.getBytes();
                for (Object data : dataList) {
                    byte[] val = null;
                    if (mineSerializer != null) {
                        val = mineSerializer.encode(data);
                    }
                    redisConnection.rPush(serialize, val);
                }
                redisConnection.close();
                return null;
            }
        });
    }

    @Override
    public void sendData(String queueName, Object data) throws InterruptedException {
        this.redisListCommit.submit(queueName, data);
    }

    @Override
    public String getType() {
        return RedisTemplate.class.getSimpleName();
    }

    @Override
    public Commit getCommit() {
        return this.redisListCommit;
    }
}
